package org.KeyCrafter.server.handler;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.redis.ErrorRedisMessage;
import org.KeyCrafter.cmd.Command;
import org.KeyCrafter.server.session.ClientSession;
import org.KeyCrafter.server.ServerContext;
import org.KeyCrafter.util.ReplyUtil;
import org.KeyCrafter.util.SharedReply;

import static org.KeyCrafter.server.ServerConstants.*;
import static org.KeyCrafter.server.aof.AofAttribute.KeyCrafter_AOF_OFF;
import static org.KeyCrafter.server.AppendOnlyFile.feedAppendOnlyFile;
import static org.KeyCrafter.server.session.ClientSession.KeyCrafter_FORCE_AOF;
import static org.KeyCrafter.util.TimeUtil.microTime;

/**
 * 客户端 handler
 * 重写 channelRead0方法，处理接收到的消息
 */

@ChannelHandler.Sharable
// 查询处理
public class CommandHandler extends SimpleChannelInboundHandler<ClientSession> {

    // 命令调用标志
    private static final int KeyCrafter_CALL_NONE = 0;
    private static final int KeyCrafter_CALL_SLOWLOG = 1;
    private static final int KeyCrafter_CALL_STATS = 2;
    private static final int KeyCrafter_CALL_PROPAGATE = 4;
    private static final int KeyCrafter_CALL_FULL = (KeyCrafter_CALL_SLOWLOG | KeyCrafter_CALL_STATS | KeyCrafter_CALL_PROPAGATE);

    /**
     * 1. Redis接收到一个写操作的命令。
     * 2. Redis将该命令追加到AOF缓冲区中。
     * 3. 定期或根据配置的条件，Redis将AOF缓冲区中的命令写入AOF文件中。
     * 4. AOF文件中的命令被写入磁盘，实现持久化。
     */
    // 命令传播标志
    private static final int KeyCrafter_PROPAGATE_NONE = 0;
    private static final int KeyCrafter_PROPAGATE_AOF = 1;
    private static final ServerContext sc = ServerContext.DEFAULT;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ClientSession c) throws Exception {
        if (processCommand(c) == KeyCrafter_OK) {
            // do something
            // ...

        }
        ctx.fireChannelRead(c);

    }

    /*
     * 这个函数执行时，我们已经读入了一个完整的命令到客户端，
     * 这个函数负责执行这个命令，
     * 或者服务器准备从客户端中进行一次读取。
     *
     * 如果这个函数返回 1 ，那么表示客户端在执行命令之后仍然存在，
     * 调用者可以继续执行其他操作。
     * 否则，如果这个函数返回 0 ，那么表示客户端已经被销毁。
     */
    private int processCommand(ClientSession c) {
        // 特别处理 quit 命令
        if (c.argv(0).toLowerCase().equals("quit")) {
            c.setReplyMessage(SharedReply.ok);
            // 关闭
            c.channel().close();
            return KeyCrafter_ERR;
        }
        //查找命令，并进行命令合法性检查，以及命令参数个数检查
        Command cmd = sc.lookupCommand(c.argv(0));

        if (cmd == null) {
            // 没找到指定的命令
            ErrorRedisMessage errMsg = new ErrorRedisMessage(String.format("unknown command '%s'", c.argv(0)));
            ReplyUtil.addReply(c,errMsg);
            return KeyCrafter_OK;
        } else if ((cmd.arity() > 0 && cmd.arity() != c.argc()) ||
                (c.argc() < -cmd.arity())) {
            // 参数个数错误
            ErrorRedisMessage errMsg = new ErrorRedisMessage(String.format("wrong number of arguments for '%s' command", cmd.name()));
            ReplyUtil.addReply(c,errMsg);
            return KeyCrafter_OK;
        }
        // 执行命令
        c.setCmd(cmd);
        call(c,KeyCrafter_CALL_FULL);
        return KeyCrafter_OK;
    }

    // 调用命令的实现函数，执行命令
    private void call(ClientSession c, int flags) {
        // 保留旧 dirty 计数器值
        long dirty = sc.getDirty();
        // 计算命令开始执行的时间
        long start = microTime();
        // 执行实现函数
        c.cmd().proc(c);
        // 计算命令执行耗费的时间
        long duration = microTime() - start;
        // 计算命令执行之后的 dirty 值
        dirty = sc.getDirty() - dirty;
        // 更新命令的统计信息
        if ((flags & KeyCrafter_CALL_STATS) != 0) {
            c.cmd().accumulateTime(duration);
            c.cmd().callsIncrement();
        }

        // 将命令复制到 AOF 节点
        if ((flags & KeyCrafter_CALL_PROPAGATE) !=0) {
            int pFlags = KeyCrafter_PROPAGATE_NONE; // 传播标志
            // 客户端强制 AOF 传播
            if ((c.getFlags() & KeyCrafter_FORCE_AOF) != 0) {
                pFlags |= KeyCrafter_PROPAGATE_AOF;
            }
            // 如果数据库有被修改，那么启用 AOF 传播
            if (dirty > 0) {
                pFlags |= KeyCrafter_PROPAGATE_AOF;
            }
            // 判断是否启用了传播
            if (pFlags != KeyCrafter_PROPAGATE_NONE) {
                propagate(c.cmd(),c.db().id(),c.getQueryBuf(),pFlags);
            }
        }
    }

    private void propagate(Command cmd, int dbId, ByteBuf queryBuf, int flags) {
        // 传播到 AOF
        if (sc.getAOFAttribute().getState() != KeyCrafter_AOF_OFF
                && (flags & KeyCrafter_PROPAGATE_AOF) != 0) {
            feedAppendOnlyFile(cmd,dbId,queryBuf);
        }
    }
}